package com.jch.fs.server;

import com.sun.xml.internal.messaging.saaj.util.ByteInputStream;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.stream.ChunkedStream;
import io.netty.util.ReferenceCountUtil;

import java.text.SimpleDateFormat;
import java.util.Date;

public class FsServerChunkHandler extends ChannelOutboundHandlerAdapter {

    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");

    static String now() {
        return sdf.format(new Date());
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        // 验证消息
        if (!(msg instanceof ByteBuf)) {
            super.write(ctx, msg, promise);
            return;
        }
        ByteBuf buf = (ByteBuf) msg;
        byte[] data = getData(buf); // 获取对应的字节数组
        // 写入流
        ByteInputStream in = new ByteInputStream();
        in.setBuf(data);

        // ChunkedStream 实现了 ChunkedInput接口【实现该接口的类都具有有关分块的功能】
        // 消息分块 | 每10个字节为一块
        ChunkedStream stream = new ChunkedStream(in, 10);

        // 消息传输通知
        ChannelProgressivePromise progressivePromise = ctx.channel().newProgressivePromise();
        // 设置消息传输监听
        progressivePromise.addListener(new ChannelProgressiveFutureListener() {
            // 监听操作进行
            @Override
            public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) throws Exception {
            }
            // 监听操作完成
            @Override
            public void operationComplete(ChannelProgressiveFuture future) throws Exception {
                if (future.isSuccess()) {
                    System.out.println(now() + " send message success.");
                    promise.setSuccess();
                } else {
                    System.err.println(now() + " send message failure：" + future.cause().getMessage());
                    promise.setFailure(future.cause());
                }
            }
        });
        // 释放消息对象引用
        ReferenceCountUtil.release(msg);
        // 写入分段信息以及特殊监听
        ctx.write(stream, progressivePromise);
    }

    private byte[] getData(ByteBuf buf) {
        if (buf.hasArray()) {
            return buf.array().clone();
        }
        byte[] data = new byte[buf.readableBytes() - 1];
        buf.readBytes(data);
        return data;
    }
}
